package per.zqx.rpc;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import per.zqx.utils.RabbitUtils;

import java.io.Closeable;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 * RpcClient，客户端，远程调用服务端计算斐波那契数列
 *
 * @author Sa  2021-06-02 21:49
 */
public class RpcClient implements Closeable {
    private Connection connection;
    private Channel channel;
    static final String QUEUE_NAME = "rpc_queue_test";

    // 构造器拿到连接和信道
    public RpcClient() throws IOException, TimeoutException {
        connection = RabbitUtils.getConnection();
        channel = connection.createChannel();
    }

    public static void main(String[] args) {
        // 创建调用者，调用服务计算斐波那契
        try (final RpcClient rpcClient = new RpcClient()){
            for (int i = 0; i < 32; i++) {
                final String i_str = Integer.toString(i);
                System.out.println("Send requesting fib(" + i_str + ")");
                String resp = rpcClient.call(i_str);
                System.out.println("Result is -> " + resp);
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }
    // 调用方法
    public String call(String message) throws IOException, InterruptedException {
        // 产生一个唯一的 correlationId，服务端要凭此返还结果
        final String correlationId = UUID.randomUUID().toString();
        // 临时队列,服务端将结果返还到这个队列
        final String replyQueueName = channel.queueDeclare().getQueue();
        // 构建资源
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .correlationId(correlationId)
                .replyTo(replyQueueName)
                .build();
        // 发布消息
        channel.basicPublish("",QUEUE_NAME,props,message.getBytes(StandardCharsets.UTF_8));
        // 容量为1的阻塞队列，存储结果，保证一次只有一个任务
        final ArrayBlockingQueue<Object> resp = new ArrayBlockingQueue<>(1);
        // 守着消息队列，等服务端的结果消息
        final String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
            // 如果 correlationId一样，放入到阻塞队列
            if (delivery.getProperties().getCorrelationId().equals(correlationId)) {
                resp.offer(new String(delivery.getBody(), "utf-8"));
            }
        }, consumerTag -> {
        });
        // 从阻塞队列拿到结果
        String result = (String) resp.take();
        // 任务已经结束，消费者取消订阅
        channel.basicCancel(ctag);
        return result;
    }

    @Override
    public void close() throws IOException {
        connection.close();
    }
}
